package org.NooLab.texx.tcp;

import java.util.Observable;
import java.util.Observer;

import org.NooLab.texx.TexxFactory;
import org.NooLab.texx.properties.TexxProperties;
import org.NooLab.tikaserver.intf.TikaSrvCmdPkgIntf;
import org.NooLab.tikaserver.intf.TikaSrvDataTransferPkgIntf;
import org.NooLab.tikaserver.tcp.agents.TikaSrvCmdPkg;
import org.NooLab.tikaserver.tcp.agents.TikaSrvDataTransferPkg;
import org.NooLab.utilities.logging.PrintLog;
import org.NooLab.utilities.objects.StringedObjects;

 

/**
 * 
 * 
 * 
 * 
 *
 */
public class TikaServerConnection extends Observable implements Observer {

	public static final int _TCP_CLIENT_WAITING_MODE_IMMEDIATE  = 1 ;
	public static final int _TCP_CLIENT_WAITING_MODE_COMPLETION = 2 ;
	
	/** note that the clientAgent contains both a client AND a server!! */
	TikaSrvClientAgent clientAgent;
	TikaDeliveryQueue tikaDeliveryQueue ;

	TikaSrvDataTransferPkgIntf dataTransferPkg;
	
	TexxFactory texxFactory;
	private TexxProperties txxProperties;

	// this should be taken dynamically, if there is the info from CLI argument like "-tika: in-7201,out-7202"
	int listenPort = 7209; // default
	int sendPort = 7210;
	Observer observerForTsc=null;
	
	int waitingMode = -1;
	private boolean msgRouted;

	StringedObjects sob = new StringedObjects();
	PrintLog out = new PrintLog(2,true);
	
	// ========================================================================
	public TikaServerConnection( Observer obs, TexxFactory factory, boolean xping){
		
		observerForTsc = obs ;
		addObserver(obs);
		 
		// getting the ports from configuration
		texxFactory = factory;
		
		txxProperties = texxFactory.getTxxProperties();
		
		if ((txxProperties.getTikaPorts()!=null) && (txxProperties.getTikaPorts()[0]>1100)){
			listenPort = txxProperties.getTikaPorts()[0]; // // overwriting default of 7209
			sendPort   = txxProperties.getTikaPorts()[1]; // where tika server is listening
		}
		
		start(); // also creates tikaDeliveryQueue 
		
		if (xping){
			new XPing(xping);  
		}
	}

	// this is being called by 
	private TikaServerConnection( Observer obs, TexxFactory factory ){
		
		observerForTsc = obs ;
		addObserver(obs);
	
		// getting the ports from configuration
		texxFactory = factory;
		txxProperties = texxFactory.getTxxProperties();
		if ((txxProperties!=null) && (txxProperties.getTikaPorts()!=null) && (txxProperties.getTikaPorts()[0]>1100)){
			listenPort = txxProperties.getTikaPorts()[0]; // // overwriting default of 7209
			sendPort   = txxProperties.getTikaPorts()[1]; // where tika server is listening
		}

	}

	// ========================================================================
	
	
	private void start() {
		 
		// the tika queue stuff
		
		tikaDeliveryQueue = new TikaDeliveryQueue( this ) ;	
		tikaDeliveryQueue.start();
		 
	}
	
	public void openConnection(){
		
		clientAgent = new TikaSrvClientAgent(this,listenPort, sendPort );
		 
	}
	
	class Wait implements Runnable{
		
		String suid="";
		Thread itccThrd;
		
		public Wait(String suid, int separateThread) {
			this.suid = suid ;
			
			if (separateThread>=1){
				itccThrd = new Thread(this,"itccThrd") ;
				itccThrd.start();
			}else{
				 waiting();
			}
		}
	
	
		private void waiting(){
			while (itemIsCompleted(suid)==false){
				out.delay(10);
			} 
		}
		
		@Override
		public void run() {
			
			waiting();
			
			int state = tikaDeliveryQueue.getStatusByGuid(suid);
			out.print(2, "waiting completed with status code : "+state) ;
		}
	
	}



	public String send(String str) throws Exception{
		String suid ;
		
		if (clientAgent==null){
			openConnection();
		}
		
		int nobs = countObservers();
		if (nobs==0){
			addObserver(observerForTsc);
		}
		// we have to prepare the specific data package
		
		TikaSrvDataTransferPkg datapkg = new TikaSrvDataTransferPkg(); 
		
		datapkg.setTypeOfData( TikaSrvDataTransferPkgIntf._TRANSFER_DATA_TYPE_LOCALFILE );
		datapkg.setFilename( str );
		
		// encoding it to a string before heading it over to the TCP infrastructure
		String encstr = sob.encode(datapkg);		
		msgRouted = false;
		
						out.print(3, "before sending...");
		suid = clientAgent.send( encstr );
		
		if (suid.contentEquals("-1")){
			throw(new Exception("Sending error"));
		}
						out.print(3, "after sending, suid = "+suid);
		
		boolean wasWaiting = false;
		// dependent on mode, we wait here or we return immediately
		// in the second case, the caller to TikaServerConnection must wait
						if (msgRouted==false){
							out.print(3, "now waiting...");
						}
		if ((msgRouted==false) && (waitingMode == _TCP_CLIENT_WAITING_MODE_COMPLETION)){
			wasWaiting=true;
			new Wait(suid,0);
		}
		
		// this mode needs the definition of an observer !!!
		// if none is defined, "this" always will wait ;
		if (waitingMode == _TCP_CLIENT_WAITING_MODE_IMMEDIATE){
			new Wait(suid,1);
		}
						if (wasWaiting){
							out.print(3, "TikaServerConnection, waiting over, continue!");
						}
		
		return suid;
	}
	
	// this returns from tikaDeliveryQueue = the queue awaiting and managing the returns
	@Override
	public void update(Observable sender, Object dataobj) {
	
		TikaSrvClientAgent tscAgent ;
		
		
		if (sender instanceof TikaSrvClientAgent == false){
			return;
		}
		
		tscAgent = (TikaSrvClientAgent)sender ;
		
		if ((dataobj instanceof TikaSrvDataTransferPkgIntf)==false){
			// cmd package or sth else
			out.print(3, "ping responded...");
			return;
		}
		TikaSrvDataTransferPkgIntf	transferData  = (TikaSrvDataTransferPkgIntf)dataobj ;
 		
		String guid = transferData.getGuid();
		tikaDeliveryQueue.removeItem(guid);
		
		String text = transferData.getResultText() ;
		
		
		int confcode = transferData.getConfirmationCode();

		// usually we get a 1 for confirming that sth has arrived (text still empty), 
		// and <0 for some problem (text will be empty then as a final result)
		if (confcode==1){
			String encstr = sob.encode(transferData);
			// no problem here, since the text is still empty
			tikaDeliveryQueue.add(guid,encstr) ;  
		}
		if ((confcode>5) || (confcode<0)){
			out.print(4, "message received by TSC, routing to WDO...");
			msgRouted=true;
			setChanged();
			notifyObservers(text); // -> WorkDatObject
		}
	}


	// --------------------------------------------------------------
	class XPing{
		
		int xServerPort ;
		boolean restart;
		
		// ................................................
		public XPing(boolean restart){
			xServerPort = sendPort;
			this.restart = restart;
			check();
		}
		// ................................................		
		private void check(){
	
			boolean isDown=true;
			
			int z=0;
			
			while ((isDown) && (z<3)){
				
				try{
					
					isDown = tcpServerPing(xServerPort) == false;
	
					if (isDown){
						z++;
						if ((isDown) && (restart)){
							out.printErr(1, "trying to restart tika server ("+z+")...");
							texxFactory.restartExternalServer(xServerPort,listenPort);
							out.delay(3000);
						}
						if (restart==false){
							break;
						}
	
					}
				}catch(Exception e){
					
			}
	
			} // ->
			
			z=0;
		}
		
		
		private boolean tcpServerPing(int srvport) {
			boolean pingOk = false;
	
			try {
	
				ping();
	
				pingOk = true;
	
			} catch (Exception e) { // educated silence...
				out.print(2, "tika server does not respond...");
			}
	
			return pingOk;
		}
	}



	public void cancelTaskByData(String filename){
		
	}

	public void cancelTaskByGuid(String guidStr){
		
	}

	protected boolean itemIsCompleted(String suid){
		boolean rB=false;
		int state = -99 ;
		
		state = tikaDeliveryQueue.getStatusByGuid(suid);
		
		rB =  (state<0) || (state==Delivery._STATUS_COMPLETED) ;
		
		return rB;
	}
	
	public int getWaitingMode() {
		return waitingMode;
	}


	public void setWaitingMode(int waitingMode) {
		this.waitingMode = waitingMode;
	}
	
	
	public int getListenPort() {
		return listenPort;
	}


	public void setListenPort(int listenPort) {
		this.listenPort = listenPort;
	}


	public int getSendPort() {
		return sendPort;
	}


	public void setSendPort(int sendPort) {
		this.sendPort = sendPort;
	}


	public int ping() {
		 
		int result = -1;
		
		TikaSrvCmdPkgIntf cmd = new TikaSrvCmdPkg();
		cmd.setPing() ;
		
		String encstr = sob.encode(cmd);		
		
		out.print(4, "before sending ping...");
		
		TikaServerConnection adhocTsc = null;
		
		try{
			
			adhocTsc = new TikaServerConnection(observerForTsc,texxFactory);	
			adhocTsc.openConnection();
			
			if (adhocTsc.clientAgent!=null){
				// result = 0 if ping would be possible for our side,
				// if it fails during operation it will raise an exception
				result=1;
			}
			
		}catch(Exception e){
			
		}
		if ((adhocTsc!=null) && (adhocTsc.clientAgent!=null)){
			result=0;
			adhocTsc.clientAgent.send( encstr );
			out.print(3, "ping was successful ...");	
			adhocTsc.disconnect();
			// reassigning observer
			addObserver(observerForTsc);
		}
		return result;
	}

	
	public void stop() {
		 
		TikaSrvCmdPkgIntf cmd = new TikaSrvCmdPkg();
		cmd.setStop() ;
		
		String encstr = sob.encode(cmd);		
		
		out.print(2, "before sending stop...");
		clientAgent.send( encstr );

	}


	public void disconnect() {
		disconnect(true) ;
	}
	public void disconnect(boolean removeObservers) {
		 
		if (removeObservers){
			this.deleteObserver(observerForTsc);
		}
		
		clientAgent.close(1) ;
	}
}
